package com.hjy.springbootinit.mq;

import com.rabbitmq.client.*;

/**
 * @version 1.0
 * @Author HJY_Maker
 * @Date 2024/5/21 星期二 14:33
 */

/**
 * 两个消费者 消费消息
 */
public class MultiConsumer {
    private static final String TASK_QUEUE_NAME = "multi_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        final Connection connection = factory.newConnection();

        //控制单个消费者的处理任务积压数  每个消费者最多只能处理1个监听
        for (int i = 0; i < 2; i++) {

            //定义如何处理消息
            int finalI = i;
            final Channel channel = connection.createChannel();
            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
            channel.basicQos(1);
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                try {
                    //处理工作
                    System.out.println(" [x] Received '" + finalI + message + "'");
                    //停20秒模拟机器处理能力有限
                    Thread.sleep(20000);
                    //第二个参数表示是否一次性确认所有的历史消息直到当前这个
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                } catch (InterruptedException e) {
                    e.getStackTrace();
                    //第二个参数跟上面的一样意思,第三个参数是 是否重新发回消息队列(相当于是否重新把物品放到快递站)
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false,false);
                } finally {
                    System.out.println(" [x] Done" + finalI);
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                }
            };
            //开启消费监听
            channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {
            });
        }
    }

}
